package cn.gs.util.set;

import java.io.File;
import java.io.IOException;
import java.io.Serializable;
import java.util.AbstractQueue;
import java.util.Iterator;
import java.util.concurrent.atomic.AtomicLong;


import com.sleepycat.bind.EntryBinding;
import com.sleepycat.bind.serial.SerialBinding;
import com.sleepycat.bind.serial.StoredClassCatalog;
import com.sleepycat.bind.tuple.TupleBinding;
import com.sleepycat.collections.StoredMap;
import com.sleepycat.collections.StoredSortedMap;
import com.sleepycat.je.Database;
import com.sleepycat.je.DatabaseConfig;
import com.sleepycat.je.DatabaseException;
import com.sleepycat.je.DatabaseExistsException;
import com.sleepycat.je.DatabaseNotFoundException;
import com.sleepycat.je.EnvironmentConfig;

/**
 * 持久化队列,基于BDB实现,也继承Queue,以及可以序列化.但不等同于Queue的时,不再使用后需要关闭
 * 相比一般的内存Queue,插入和获取值需要多消耗一定的时间
 * 这里为什么是继承AbstractQueue而不是实现Queue接口,是因为只要实现offer,peek,poll几个方法即可,
 * 其他如remove,addAll,AbstractQueue会基于这几个方法去实现
 *
 * @contributor guoyun
 *
 * @param <E>
 */
public class BdbPersistentQueue<E extends Serializable> extends AbstractQueue<E> implements
        Serializable {

    private static final long serialVersionUID = 3427799316155220967L;
    private transient BDBMap<Long, E> queueMap;   // 持久化Map,Key为指针位置,Value为值,无需序列化   
    private AtomicLong headIndex;                   // 头部指针  
    private AtomicLong tailIndex;                   // 尾部指针  
    private transient E peekItem = null;              // 当前获取的值  

    /**
     * 构造函数,传入BDB数据库位置和名字,自己创建数据库
     *
     * @param dbDir
     * @param dbName
     * @param valueClass
     */
    public BdbPersistentQueue(String dbDir, String dbName, Class<E> valueClass) {
        headIndex = new AtomicLong(0);
        tailIndex = new AtomicLong(0);
        queueMap = BDBMap.createAndBindDatabase(dbDir, dbName, Long.class, valueClass);
    }

    /**
     * 值遍历器
     */
    @Override
    public Iterator<E> iterator() {
        return queueMap.values().iterator();
    }

    /**
     * 大小
     */
    @Override
    public int size() {
        synchronized (tailIndex) {
            synchronized (headIndex) {
                return (int) (tailIndex.get() - headIndex.get());
            }
        }
    }

    /**
     * 插入值
     */
    @Override
    public boolean offer(E e) {
        synchronized (tailIndex) {
            queueMap.put(tailIndex.getAndIncrement(), e);   // 从尾部插入  
        }
        return true;
    }

    /**
     * 获取值,从头部获取
     */
    @Override
    public E peek() {
        synchronized (headIndex) {
            if (peekItem != null) {
                return peekItem;
            }
            E headItem = null;
            while (headItem == null && headIndex.get() < tailIndex.get()) { // 没有超出范围  
                headItem = queueMap.get(headIndex.get());
                if (headItem != null) {
                    peekItem = headItem;
                    continue;
                }
                headIndex.incrementAndGet();    // 头部指针后移  
            }
            return headItem;
        }
    }

    /**
     * 移出元素,移出头部元素
     */
    @Override
    public E poll() {
        synchronized (headIndex) {
            E headItem = peek();
            if (headItem != null) {
                queueMap.remove(headIndex.getAndIncrement());
                peekItem = null;
                return headItem;
            }
        }
        return null;
    }

    /**
     * 关闭,也就是关闭所是用的BDB数据库但不关闭数据库环境
     */
    public void close() {
    }

    /**
     * 清理,会清空数据库,并且删掉数据库所在目录,慎用.如果想保留数据,请调用close()
     */
    @Override
    public void clear() {
    }
}
